[ML] Ensure queued AbstractRunnables are notified when executor stops #135966

DonalEvans · 2025-10-03T19:35:26Z

AbstractProcessWorkerExecutorService.notifyQueueRunnables() was making an incorrect assumption that all AbstractRunnables that were submitted for execution would be queued as AbstractRunnables. However, PriorityProcessWorkerExecutorService wraps AbstractRunnables in OrderedRunnable before queueing them, and since OrderedRunnable is not an AbstractRunnable, these were skipped when notifyQueueRunnables() drained the queue, leading to potential hangs.

Refactor notifyQueueRunnables() to allow PriorityProcessWorkerExecutorService to notify the AbstractRunnable contained within queued OrderedRunnables
Ensure that notifyQueueRunnables() is called and the executor marked as shut down if an exception is thrown from start()
Add unit tests

Closes #134651

AbstractProcessWorkerExecutorService.notifyQueueRunnables() was making an incorrect assumption that all AbstractRunnables that were submitted for execution would be queued as AbstractRunnables. However, PriorityProcessWorkerExecutorService wraps AbstractRunnables in OrderedRunnable before queueing them, and since OrderedRunnable is not an AbstractRunnable, these were skipped when notifyQueueRunnables() drained the queue, leading to potential hangs. - Refactor notifyQueueRunnables() to allow PriorityProcessWorkerExecutorService to notify the AbstractRunnable contained within queued OrderedRunnables - Ensure that notifyQueueRunnables() is called and the executor marked as shut down if an exception is thrown from start() - Add unit tests Closes elastic#134651

elasticsearchmachine · 2025-10-03T19:35:52Z

Pinging @elastic/ml-core (Team:ML)

elasticsearchmachine · 2025-10-03T19:35:53Z

Hi @DonalEvans, I've created a changelog YAML for you.

jonathan-buttner

Great work, left one suggestion

jonathan-buttner · 2025-10-06T15:24:58Z

...c/main/java/org/elasticsearch/xpack/ml/job/process/AbstractProcessWorkerExecutorService.java

+        }
+    }
+
+    protected abstract void notifyIfAbstractRunnable(T runnable, Exception ex, String msg);


What do you think about having AbstractProcessWorkerExecutorService contain the logic for notifyIfAbstractRunnable and relying on an abstract method like getAsAbstractRunnable which either returns the AbstractRunnable or null. That way child classes don't need to call back into the parent to do the notification, they would only return the abstract runnable if it was one.

Something like:

protected abstract AbstractRunnable getAsAbstractRunnable(T runnable); private void notifyIfAbstractRunnable(T runnable, Exception ex, String msg) { var abstractRunnable = getAsAbstractRunnable(runnable); if (abstractRunnable != null) { notifyAbstractRunnable(ex, msg, abstractRunnable); } }

Then PriorityProcessWorkerExecutorService would have something like:

@Override protected AbstractRunnable getAsAbstractRunnable(OrderedRunnable orderedRunnable, Exception ex, String msg) { // The runnable contained within OrderedRunnable is always an AbstractRunnable, so no need to check the type return orderedRunnable.runnable(); }

Good idea, this is better

davidkyle · 2025-10-07T14:50:18Z

.../java/org/elasticsearch/xpack/ml/inference/pytorch/PriorityProcessWorkerExecutorService.java

     * most cases will be the insertion order
     */
-    public record OrderedRunnable(RequestPriority priority, long tieBreaker, Runnable runnable)
+    protected record OrderedRunnable(RequestPriority priority, long tieBreaker, AbstractRunnable runnable)


Another option is make OrderedRunnable extend AbstractRunnable and implement the required methods as calling the runnable parameter

This solution adds more code than the original one, but I think it might be better, since it simplifies the logic for determining what we need to notify.

…elastic#135966) AbstractProcessWorkerExecutorService.notifyQueueRunnables() was making an incorrect assumption that all AbstractRunnables that were submitted for execution would be queued as AbstractRunnables. However, PriorityProcessWorkerExecutorService wraps AbstractRunnables in OrderedRunnable before queueing them, and since OrderedRunnable is not an AbstractRunnable, these were skipped when notifyQueueRunnables() drained the queue, leading to potential hangs. - Make OrderedRunnable extend AbstractRunnable and pass onFailure() and onRejection() calls to the AbstractRunnable it wraps - Ensure that notifyQueueRunnables() is called and the executor marked as shut down if an exception is thrown from start() - Add unit tests - Update docs/changelog/135966.yaml Closes elastic#134651

elasticsearchmachine · 2025-10-07T18:57:01Z

💔 Backport failed

Status	Branch	Result
✅	9.2
❌	8.19	Commit could not be cherrypicked due to conflicts
❌	9.1	Commit could not be cherrypicked due to conflicts
❌	9.0	Commit could not be cherrypicked due to conflicts
❌	8.18	Commit could not be cherrypicked due to conflicts

You can use sqren/backport to manually backport by running backport --upstream elastic/elasticsearch --pr 135966

…elastic#135966) AbstractProcessWorkerExecutorService.notifyQueueRunnables() was making an incorrect assumption that all AbstractRunnables that were submitted for execution would be queued as AbstractRunnables. However, PriorityProcessWorkerExecutorService wraps AbstractRunnables in OrderedRunnable before queueing them, and since OrderedRunnable is not an AbstractRunnable, these were skipped when notifyQueueRunnables() drained the queue, leading to potential hangs. - Make OrderedRunnable extend AbstractRunnable and pass onFailure() and onRejection() calls to the AbstractRunnable it wraps - Ensure that notifyQueueRunnables() is called and the executor marked as shut down if an exception is thrown from start() - Add unit tests - Update docs/changelog/135966.yaml Closes elastic#134651 (cherry picked from commit d3d013e)

DonalEvans · 2025-10-07T19:32:51Z

💚 All backports created successfully

Status	Branch	Result
✅	9.1
✅	9.0
✅	8.19
✅	8.18

Questions ?

Please refer to the Backport tool documentation

…elastic#135966) AbstractProcessWorkerExecutorService.notifyQueueRunnables() was making an incorrect assumption that all AbstractRunnables that were submitted for execution would be queued as AbstractRunnables. However, PriorityProcessWorkerExecutorService wraps AbstractRunnables in OrderedRunnable before queueing them, and since OrderedRunnable is not an AbstractRunnable, these were skipped when notifyQueueRunnables() drained the queue, leading to potential hangs. - Make OrderedRunnable extend AbstractRunnable and pass onFailure() and onRejection() calls to the AbstractRunnable it wraps - Ensure that notifyQueueRunnables() is called and the executor marked as shut down if an exception is thrown from start() - Add unit tests - Update docs/changelog/135966.yaml Closes elastic#134651 (cherry picked from commit d3d013e)

…#135966) (#136122) AbstractProcessWorkerExecutorService.notifyQueueRunnables() was making an incorrect assumption that all AbstractRunnables that were submitted for execution would be queued as AbstractRunnables. However, PriorityProcessWorkerExecutorService wraps AbstractRunnables in OrderedRunnable before queueing them, and since OrderedRunnable is not an AbstractRunnable, these were skipped when notifyQueueRunnables() drained the queue, leading to potential hangs. - Make OrderedRunnable extend AbstractRunnable and pass onFailure() and onRejection() calls to the AbstractRunnable it wraps - Ensure that notifyQueueRunnables() is called and the executor marked as shut down if an exception is thrown from start() - Add unit tests - Update docs/changelog/135966.yaml Closes #134651

…#135966) (#136126) AbstractProcessWorkerExecutorService.notifyQueueRunnables() was making an incorrect assumption that all AbstractRunnables that were submitted for execution would be queued as AbstractRunnables. However, PriorityProcessWorkerExecutorService wraps AbstractRunnables in OrderedRunnable before queueing them, and since OrderedRunnable is not an AbstractRunnable, these were skipped when notifyQueueRunnables() drained the queue, leading to potential hangs. - Make OrderedRunnable extend AbstractRunnable and pass onFailure() and onRejection() calls to the AbstractRunnable it wraps - Ensure that notifyQueueRunnables() is called and the executor marked as shut down if an exception is thrown from start() - Add unit tests - Update docs/changelog/135966.yaml Closes #134651 (cherry picked from commit d3d013e)

…#135966) (#136128) AbstractProcessWorkerExecutorService.notifyQueueRunnables() was making an incorrect assumption that all AbstractRunnables that were submitted for execution would be queued as AbstractRunnables. However, PriorityProcessWorkerExecutorService wraps AbstractRunnables in OrderedRunnable before queueing them, and since OrderedRunnable is not an AbstractRunnable, these were skipped when notifyQueueRunnables() drained the queue, leading to potential hangs. - Make OrderedRunnable extend AbstractRunnable and pass onFailure() and onRejection() calls to the AbstractRunnable it wraps - Ensure that notifyQueueRunnables() is called and the executor marked as shut down if an exception is thrown from start() - Add unit tests - Update docs/changelog/135966.yaml Closes #134651 (cherry picked from commit d3d013e)

…#135966) (#136125) AbstractProcessWorkerExecutorService.notifyQueueRunnables() was making an incorrect assumption that all AbstractRunnables that were submitted for execution would be queued as AbstractRunnables. However, PriorityProcessWorkerExecutorService wraps AbstractRunnables in OrderedRunnable before queueing them, and since OrderedRunnable is not an AbstractRunnable, these were skipped when notifyQueueRunnables() drained the queue, leading to potential hangs. - Make OrderedRunnable extend AbstractRunnable and pass onFailure() and onRejection() calls to the AbstractRunnable it wraps - Ensure that notifyQueueRunnables() is called and the executor marked as shut down if an exception is thrown from start() - Add unit tests - Update docs/changelog/135966.yaml Closes #134651 (cherry picked from commit d3d013e)

…#135966) (#136127) AbstractProcessWorkerExecutorService.notifyQueueRunnables() was making an incorrect assumption that all AbstractRunnables that were submitted for execution would be queued as AbstractRunnables. However, PriorityProcessWorkerExecutorService wraps AbstractRunnables in OrderedRunnable before queueing them, and since OrderedRunnable is not an AbstractRunnable, these were skipped when notifyQueueRunnables() drained the queue, leading to potential hangs. - Make OrderedRunnable extend AbstractRunnable and pass onFailure() and onRejection() calls to the AbstractRunnable it wraps - Ensure that notifyQueueRunnables() is called and the executor marked as shut down if an exception is thrown from start() - Add unit tests - Update docs/changelog/135966.yaml Closes #134651 (cherry picked from commit d3d013e)

DonalEvans added >bug :ml Machine learning Team:ML Meta label for the ML team auto-backport Automatically create backport pull requests when merged v8.19.6 v9.1.6 v9.0.9 v8.18.9 v9.2.1 v9.3.0 labels Oct 3, 2025

DonalEvans and others added 2 commits October 3, 2025 12:35

Update docs/changelog/135966.yaml

eb9f19b

Merge branch 'main' into 134651-reindex-hang

83fefe2

jonathan-buttner reviewed Oct 6, 2025

View reviewed changes

davidkyle reviewed Oct 7, 2025

View reviewed changes

DonalEvans added 2 commits October 7, 2025 08:56

Make OrderedRunnable extend AbstractRunnable

2767061

Merge branch 'main' into 134651-reindex-hang

4e09877

DonalEvans added the v9.2.0 label Oct 7, 2025

DonalEvans requested review from davidkyle and jonathan-buttner October 7, 2025 17:49

jonathan-buttner approved these changes Oct 7, 2025

View reviewed changes

DonalEvans merged commit d3d013e into elastic:main Oct 7, 2025
34 checks passed

DonalEvans mentioned this pull request Oct 7, 2025

[9.2] [ML] Ensure queued AbstractRunnables are notified when executor stops (#135966) #136122

Merged

elasticsearchmachine added the backport pending label Oct 7, 2025

DonalEvans mentioned this pull request Oct 7, 2025

[9.1] [ML] Ensure queued AbstractRunnables are notified when executor stops (#135966) #136125

Merged

DonalEvans deleted the 134651-reindex-hang branch October 8, 2025 20:59

DonalEvans removed the backport pending label Oct 13, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[ML] Ensure queued AbstractRunnables are notified when executor stops #135966

[ML] Ensure queued AbstractRunnables are notified when executor stops #135966

Uh oh!

DonalEvans commented Oct 3, 2025

Uh oh!

elasticsearchmachine commented Oct 3, 2025

Uh oh!

elasticsearchmachine commented Oct 3, 2025

Uh oh!

jonathan-buttner left a comment

Uh oh!

jonathan-buttner Oct 6, 2025

Uh oh!

DonalEvans Oct 7, 2025

Uh oh!

davidkyle Oct 7, 2025

Uh oh!

DonalEvans Oct 7, 2025

Uh oh!

Uh oh!

elasticsearchmachine commented Oct 7, 2025

Uh oh!

DonalEvans commented Oct 7, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

[ML] Ensure queued AbstractRunnables are notified when executor stops #135966

[ML] Ensure queued AbstractRunnables are notified when executor stops #135966

Uh oh!

Conversation

DonalEvans commented Oct 3, 2025

Uh oh!

elasticsearchmachine commented Oct 3, 2025

Uh oh!

elasticsearchmachine commented Oct 3, 2025

Uh oh!

jonathan-buttner left a comment

Choose a reason for hiding this comment

Uh oh!

jonathan-buttner Oct 6, 2025

Choose a reason for hiding this comment

Uh oh!

DonalEvans Oct 7, 2025

Choose a reason for hiding this comment

Uh oh!

davidkyle Oct 7, 2025

Choose a reason for hiding this comment

Uh oh!

DonalEvans Oct 7, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

elasticsearchmachine commented Oct 7, 2025

💔 Backport failed

Uh oh!

DonalEvans commented Oct 7, 2025

💚 All backports created successfully

Questions ?

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants